import base64
import hashlib
import json
import logging
import math
import os
import random
import re
import shutil
from datetime import date, datetime, timedelta
from time import sleep
from typing import Any, List, Optional, Union

import httpx
from dotenv import load_dotenv
from fastapi import Request, UploadFile
from markupsafe import Markup, escape
from PIL import Image, ImageOps, UnidentifiedImageError
from sqlalchemy import (
    Index, asc, cast, delete, desc, func, select, String, DateTime
)
from sqlalchemy.exc import IntegrityError
from starlette.datastructures import URL

from core.database import DBConnect, db_session, MySQLCharsetMixin
from core.models import (
    BoardNew, Config, Member, Memo, UniqId, Visit, WriteBaseModel
)
from core.plugin import get_admin_menu_id_by_path

load_dotenv()

# 全球变量的声明(global variables)
CAPTCHA_PATH = "lib/captcha/templates"
EDITOR_PATH = "lib/editor/templates"


#动态模型缓存：检查是否已创建模型，仅在未创建模型时才创建新模型的方法。
#为此，您可以使用简单的全局字典来跟踪已创建的模型。
_created_models = {}

#  创建动态公告栏模型
def dynamic_create_write_table(
        table_name: str,
        create_table: bool = False,
    ) -> WriteBaseModel:
    '''
从WriteBaseModel复制公告栏表结构并将其创建为动态模型的函数
在参数的table_name中，只接受除table_prefix+“write_”之外的表名。
Create Dynamic Write Table Model from WriteBaseModel
    '''
    # 返回已创建的模型
    if table_name in _created_models:
        return _created_models[table_name]

    if isinstance(table_name, int):
        table_name = str(table_name)

    class_name = "Write" + table_name.capitalize()
    db_connect = DBConnect()
    DynamicModel = type(
        class_name,
        (WriteBaseModel,),
        {
            "__tablename__": db_connect.table_prefix + 'write_' + table_name,
            "__table_args__": (
                Index(f'idx_wr_num_reply_{table_name}', 'wr_num', 'wr_reply'),
                Index(f'idex_wr_is_comment_{table_name}', 'wr_is_comment'),
                {
                    "extend_existing": True,
                    **MySQLCharsetMixin().__table_args__
                },
            ),
        }
    )
    #  添加公告栏时只创建一次表格
    if create_table:
        DynamicModel.__table__.create(bind=db_connect.engine, checkfirst=True)
    # 缓存创建的模型   #缓存创建的模型
    _created_models[table_name] = DynamicModel
    return DynamicModel


def session_member_key(request: Request, member: Member):
    '''
    生成并返回成员的唯一键以存储在会话中的函数  创建并返回要保存在会话中的成员的唯一密钥的函数
    '''
    ss_mb_key = hashlib.md5(
        (member.mb_datetime.strftime(format="%Y-%m-%d %H:%M:%S")
         + get_client_ip(request)
         + request.headers.get('User-Agent')).encode()).hexdigest()
    return ss_mb_key


def get_admin_menus():
    '''
    1, 2由步骤分隔的管理器菜单 json 如果有一个文件 load 要返回的函数
    '''
    files = [
        "admin/admin_menu_bbs.json",
        "admin/admin_menu_shop.json",
        "admin/admin_menu_sms.json"
    ]
    menus = {}
    for file_path in files:
        if os.path.exists(file_path):
            with open(file_path, "r", encoding="utf-8") as file:
                menus.update(json.load(file))
    return menus


def get_head_tail_img(directory: str, filename: str, width: int = 750) -> dict:
    """
    目录/查找与文件名对应的图像,
    如果该图像存在，则该图像的 URL并返回宽度.
    - 图像的宽度基本上是最大的 750px仅限于.

    Args:
        directory (str): 映像目录
        filename (str): 图像文件名

    Returns:
        dict: 图像的存在, 图像 URL, 图像宽度
        
    """
    img_path = os.path.join('data', directory, filename)
    img_exists = os.path.exists(img_path)
    img_width = 0

    if img_exists:
        try:
            with Image.open(img_path) as img_file:
                img_width = min(img_file.width, width)  # 图像的宽度 750px限于
        except UnidentifiedImageError:
            print(f"Error: Cannot identify image file '{img_path}'")
            img_exists = False

    return {
        "exists": img_exists,
        "url": f"/{img_path}" if img_exists else "",
        "width": img_width
    }


def get_client_ip(request: Request) -> str:
    '''
    客户端的 IP 函数返回地址 (PHP的 $_SERVER['REMOTE_ADDR'])
    '''
    x_forwarded_for = request.headers.get("X-Forwarded-For")
    if x_forwarded_for:
        # X-Forwarded-For can be a comma-separated list of IPs.
        # The client's requested IP will be the first one.
        return x_forwarded_for.split(",")[0]
    else:
        return request.client.host


async def get_host_public_ip() -> str:
    """
    主持人的官方 IP 函数返回地址
    """
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get('https://httpbin.org/ip')
            return response.json()['origin']
        except httpx.TimeoutException:
            return "IP 调用信息失败. 我要你再试一次.."


def delete_image(directory: str, filename: str, is_delete: bool = True):
    """图像删除处理功能

    Args:
        directory (str): 路径
        filename (str): 文件名
        is_delete (bool): 是否删除. Defaults to True.
    """
    if is_delete:
        file_path = f"{directory}/{filename}"
        if os.path.exists(file_path):
            os.remove(file_path)


def save_image(directory: str, filename: str, file: UploadFile):
    """图像存储处理功能

    Args:
        directory (str): 路径
        filename (str): 文件名
        file (UploadFile): 文件 ojbect
    """
    if file and file.filename:
        with open(f"{directory}/{filename}", "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)


# Python的内置函数， list 防止名称与变量名称冲突 lst 已更改为
def get_from_list(lst, index, default=0):
    if lst is None:
        return default
    try:
        return 1 if index in lst else default
    except (TypeError, IndexError):
        return default


def select_query(request: Request, db: db_session, table_model, search_params: dict,
        same_search_fields: Optional[List[str]] = "", # 值必须完全相同，但过滤 '搜索词'
        prefix_search_fields: Optional[List[str]] = "", # 后面 %通过粘贴过滤 '搜索词%'
        default_sod: str = "asc",
        # default_sst: Optional[List[str]] = [],
        default_sst: str = "",
    ):
    config = request.state.config
    records_per_page = config.cf_page_rows

    query = select()
    # # sod如果提供, 添加基于该列的排序.
    # if search_params['sst'] is not None and search_params['sst'] != "":
    #     # if search_params['sod'] == "desc":
    #     #     query = query.order_by(desc(getattr(table_model, search_params['sst'])))
    #     # else:
    #     #     query = query.order_by(asc(getattr(table_model, search_params['sst'])))
    #     if search_params.get('sod', default_sod) == "desc":  # 修改部分
    #         query = query.order_by(desc(getattr(table_model, search_params['sst'])))
    #     else:
    #         query = query.order_by(asc(getattr(table_model, search_params['sst'])))

    # 'sst' 如果未提供参数或是空字符串, default_sst我使用.
    sst = search_params.get('sst', default_sst) or default_sst
    # sod如果提供, 添加基于该列的排序.
    sod = search_params.get('sod', default_sod) or default_sod

    if sst:
        # sst 如果是一个数组, 여러 열을 기준으로 정렬을 추가합니다.
        # sst 가 배열인 경우, 添加基于多个列的排序.
        if isinstance(sst, list):
            for sort_attribute in sst:
                sort_column = getattr(table_model, sort_attribute)
                if sod == "desc":
                    query = query.order_by(desc(sort_column))
                else:
                    query = query.order_by(asc(sort_column))
        else:
            if sod == "desc":
                query = query.order_by(desc(getattr(table_model, sst)))
            else:
                query = query.order_by(asc(getattr(table_model, sst)))


    # sfl和 stx如果提供, 使用相应的列和值进行附加筛选.
    if search_params['sfl'] is not None and search_params['stx'] is not None:
        if hasattr(table_model, search_params['sfl']):  # sfl这个 Table检查它是否存在于
            # if search_params['sfl'] in ["mb_level"]:
            if search_params['sfl'] in same_search_fields:
                query = query.where(getattr(table_model, search_params['sfl']) == search_params['stx'])
            elif search_params['sfl'] in prefix_search_fields:
                query = query.where(getattr(table_model, search_params['sfl']).like(f"{search_params['stx']}%"))
            else:
                query = query.where(cast(getattr(table_model, search_params['sfl']), String).like(f"%{search_params['stx']}%"))

    # 根据页码 offset 计算
    page = search_params['current_page']
    offset = (page - 1) * records_per_page if page > 0 else 0
    # 获取最终查询结果.
    rows = db.scalars(query.add_columns(table_model).offset(offset).limit(records_per_page)).all()
    # 计算记录的总数
    total_count = db.scalar(query.add_columns(func.count()).select_from(table_model).order_by(None))

    return {
        "rows": rows,
        "total_count": total_count,
    }


def domain_mail_host(request: Request, is_at: bool = True):
    domain_host = request.base_url.hostname

    if domain_host.startswith("www."):
        domain_host = domain_host[4:]

    return f"@{domain_host}" if is_at else domain_host


def nl2br(value) -> str:
    """ \n 报告 <br> 转换为标签
    """
    return escape(value).replace('\n', Markup('<br>\n'))


def get_unique_id(request) -> Optional[str]:
    """独特的密钥生成功能
    nouboard 5的 get_uniqid

    年日期00 ~ 年日期99
    年(4) 月份(2) 工作(2) 城市(2) 分钟(2) 秒(2) 10010,000分钟 1秒(2)
    Args:
        request (Request): FastAPI Request 对象
    Returns:
        Optional[str]: 唯一名称, DB 在错误的时候 None
    """

    ip: str = get_client_ip(request)

    while True:
        current = datetime.now()
        ten_milli_sec = str(current.microsecond)[:2].zfill(2)
        key = f"{current.strftime('%Y%m%d%H%M%S')}{ten_milli_sec}"

        with DBConnect().sessionLocal() as session:
            try:
                session.add(UniqId(uq_id=key, uq_ip=ip))
                session.commit()
                return key

            except IntegrityError:
                # key 如果出现重复错误，请重试
                session.rollback()
                sleep(random.uniform(0.01, 0.02))
            except Exception as e:
                logging.log(logging.CRITICAL, 'unique table insert error', exc_info=e)
                return None


class StringEncrypt:
    def __init__(self, salt=''):
        if not salt:
            # You might want to implement your own salt generation logic here
            self.salt = "your_default_salt"
        else:
            self.salt = salt

        self.length = len(self.salt)

    def encrypt(self, str_):
        length = len(str_)
        result = ''

        for i in range(length):
            char = str_[i]
            keychar = self.salt[i % self.length]
            char = chr(ord(char) + ord(keychar))
            result += char

        result = base64.b64encode(result.encode()).decode()
        result = result.translate(str.maketrans('+/=', '._-'))

        return result

    def decrypt(self, str_):
        result = ''
        str_ = str_.translate(str.maketrans('._-', '+/='))
        str_ = base64.b64decode(str_).decode()

        length = len(str_)

        for i in range(length):
            char = str_[i]
            keychar = self.salt[i % self.length]
            char = chr(ord(char) - ord(keychar))
            result += char

        return result

# 使用示例
# enc = StringEncrypt()
# encrypted_text = enc.encrypt("hello")
# print(encrypted_text)

# decrypted_text = enc.decrypt(encrypted_text)
# print(decrypted_text)

class FileCache():
    """文件缓存类
    """
    cache_dir = os.path.join("data", "cache")
    cache_secret_key = None

    def __init__(self):
    
        os.makedirs(self.cache_dir, exist_ok=True)

    def get_cache_secret_key(self):
        """
       返回缓存密钥的函数
        """
        # 如果有缓存的值，则返回该值
        if self.cache_secret_key:
            return self.cache_secret_key

        # 通过散列服务器软件和DOCUMENT_ROOT值生成6位字符串
        server_software = os.environ.get("SERVER_SOFTWARE", "")
        document_root = os.environ.get("DOCUMENT_ROOT", "")
        combined_data = server_software + document_root
        self.cache_secret_key = hashlib.md5(combined_data.encode()).hexdigest()[:6]

        return self.cache_secret_key

    def get(self, cache_file: str):
        """
        如果有缓存的文件，则读取并返回文件
        """
        if os.path.exists(cache_file):
            with open(cache_file, "r", encoding="utf-8") as f:
                return f.read()
        return None

    def create(self, data: str, cache_file: str):
        """
        cache_file一个函数生成
        """
        with open(cache_file, "w", encoding="utf-8") as f:
            f.write(data)

    def delete(self, cache_file: str):
        """
        cache_file要删除的函数
        """
        if os.path.exists(cache_file):
            os.remove(cache_file)

    def delete_prefix(self, prefix: str):
        """
       删除以prefix开头的所有缓存文件的函数
        """
        for file in os.listdir(self.cache_dir):
            if file.startswith(prefix):
                os.remove(os.path.join(self.cache_dir, file))


def get_admin_email(request: Request):
    """返回管理员电子邮件地址的函数

    Args:
        request (Request): Request 对象

    Returns:
        str: 在环境设置中设置的经理电子邮件地址
    """
    return getattr(request.state.config, "cf_admin_email", "")


def get_admin_email_name(request: Request):
    """返回发送管理员电子邮件的名称的函数

    Args:
        request (Request): Request 对象

    Returns:
        str: 在环境设置中设置的经理电子邮件地址
    """
    return getattr(request.state.config, "cf_admin_email_name", "")


def is_none_datetime(input_date: Union[date, str]) -> bool:
    """date, datetime 这个 0001, 0000 函数来检查它是否是无效的日期等。
    0001, mysql 5.7下面 0000,
    """
    if isinstance(input_date, str):  # pymysql 图书馆是 '0000', 12月份 32当第一类错误的日期 str 返回类型.
        return True

    if input_date.strftime("%Y")[:2] == "00":
        return True

    return False


def calculator_image_resize(source_width, source_height, target_width=0, target_height=0):
    """
    维护和计算图像比率 , 如果仅输入宽度和高度之一，则计算比率
    原始图像是 target_width, target_height 如果小于 False 返回
    원본이미지가 target_width, target_height 보다 작으면 False 반환
    Args:
        source_width (int): 原始图像宽度
        source_height (int): 原始图像高度
        target_width (int): 要更改的图像的宽度. Defaults 0.
        target_height (int): 要更改的图像高度. Defaults 0.
    Returns:
        Union[bool, dict]: 要更改的图像的宽度, 高度 dict{'width': new_width, 'height': new_height} or False
    """
    # 如果小，不调整大小
    if source_width < target_width and source_height < target_height:
        return False

    if source_width > target_width and source_height > target_height:
        # 如果原始大于目标，则计算减速比
        ratio_width = target_width / source_width
        ratio_height = target_height / source_height
        min_ratio = min(ratio_width, ratio_height)
        return {'width': int(source_width * min_ratio), 'height': int(source_height * min_ratio)}

    # 维护和计算图像比率
    # 如果仅输入宽度和高度之一，则计算比率
    if target_width and not target_height:
        ratio = target_width / source_width
        new_width = target_width
        new_height = int(source_height * ratio)

    elif not target_width and target_height:
        ratio = target_height / source_height
        new_width = int(source_width * ratio)
        new_height = target_height

    else:
        return False  # 处理是否同时输入了宽度和高度

    return {'width': new_width, 'height': new_height}



def thumbnail(source_file: str, target_path: str = None, width: int = 200, height: int = 150, **kwargs) -> str:
    """我们创建一个缩略图.

    Args:
        source_file (str): 原始图像文件路径
        target_path (str, optional): 缩略图图像文件路径. Defaults to None.
        width (int, optional): 缩略图图像宽度. Defaults to 200.
        height (int, optional): 缩略图图像高度. Defaults to 150.

    Returns:
        str: 缩略图图像文件路径
    """
    try:
        source_basename = os.path.basename(source_file)
        source_path = os.path.dirname(source_file)
        target_path = target_path or source_path

        # 创建缩略图存储路径
        os.makedirs(target_path, exist_ok=True)

        # 缩略图文件路径
        thumbnail_file = os.path.join(target_path, f"thumbnail_{width}x{height}_{source_basename}")
        # 缩略图文件存在
        # 原始文件创建时间 < 创建缩略图文件的时间
        if os.path.exists(thumbnail_file):
            if os.path.getmtime(source_file) < os.path.getmtime(thumbnail_file):
                return thumbnail_file

        # 图像对象创建
        # 如果没有文件或没有图像，则会出现异常，因此不要单独检查.
        source_image = Image.open(source_file)
        source_width, source_height = source_image.size

        # 如果图像小于缩略图
        if source_width < width or source_height < height:
            # 扩展映像创建
            expanded_img = Image.new("RGB", (width, height), (255, 255, 255))
            # 将现有图像插入扩展图像的中心
            left = (width - source_width) // 2
            top = (height - source_height) // 2
            expanded_img.paste(source_image, (left, top))
            expanded_img.save(thumbnail_file)
        else:
            # 剪切并保存图像到指定的大小
            ImageOps.fit(source_image, (width, height)).save(thumbnail_file)
            # source_image.thumbnail((width, height))
            # source_image.save(thumbnail_file)

        return thumbnail_file

    except UnidentifiedImageError as e:
        print("用于创建原始图像对象的Fes : ", e)
        return ""

    except Exception as e:
        print("创建缩略图失败 : ", e)
        return ""


def get_editor_image(contents: str, view: bool = True) -> list:
    """从设计器中提取图像标记.

    Args:
        contents (str): 内容
        view (bool, optional): 视图模式是否. Defaults to True.

    Returns:
        list: 图像标签 src 属性值
    """
    if not contents:
        return []

    # contents 中间 img 标签提取
    if view:
        pattern = re.compile(r"<img([^>]*)>", re.IGNORECASE | re.DOTALL)
    else:
        pattern = re.compile(r"<img[^>]*src=[\'\"]?([^>\'\"]+[^>\'\"]+)[\'\"]?[^>]*>", re.IGNORECASE | re.DOTALL)

    matches = pattern.findall(contents)

    return matches


def extract_alt_attribute(img_tag: str) -> str:
    """alt 提取属性

    Args:
        img_tag (str): img 标签

    Returns:
        str: alt 属性值
    """
    alt_match = re.search(r'alt=[\"\']?([^\"\']*)[\"\']?', img_tag, re.IGNORECASE)
    alt = str(alt_match.group(1)) if alt_match else ''
    return alt


def cut_name(request: Request, name: str) -> str:
    """基本环境设置 > 姓名(昵称) 标记

    Args:
        request (Request): FastAPI Request
        name (str): 姓名

    Returns:
        str: 剪切名称
    """
    config = request.state.config

    if not name:
        return ''

    return name[:config.cf_cut_name] if config.cf_cut_name else name


def delete_old_records():
    """
    删除设置日期之后的数据
    """
    try:
        db = DBConnect().sessionLocal()
        config = db.scalar(select(Config))
        today = datetime.now()

        # 删除访客记录
        if config.cf_visit_del > 0:
            base_date = today - timedelta(days=config.cf_visit_del)
            if db.bind.dialect.name == "sqlite":
                visit_datetime = Visit.vi_date.concat(" ").concat(Visit.vi_time)
                concat_expr = func.strftime("%Y-%m-%d %H:%M:%S", visit_datetime)
            else:
                concat_expr = func.cast(func.concat(Visit.vi_date, " ", Visit.vi_time), DateTime)
            result = db.execute(
                delete(Visit).where(concat_expr < base_date)
            )
            print("删除访客记录的参考日 : ", base_date, f"{result.rowcount}删除枪")
            print("방문자기록 삭제 기준일 : ", base_date, f"{result.rowcount}건 삭제")

        # 删除热门搜索词
        if config.cf_popular_del > 0:
            from service.popular_service import PopularService

            base_date = today - timedelta(days=config.cf_popular_del)
            popular_service = PopularService(db)
            delete_count = popular_service.delete_populars(base_date.date())

            print("删除热门搜索词的标准日 : ", base_date, f"{delete_count}删除枪")

        # 删除最近的帖子
        if config.cf_new_del > 0:
            base_date = today - timedelta(days=config.cf_new_del)
            result = db.execute(
                delete(BoardNew).where((BoardNew.bn_datetime != None) & (BoardNew.bn_datetime < base_date))
            )
            print("最新帖子删除参考日 : ", base_date, f"{result.rowcount}删除枪")

        # 删除便笺
        if config.cf_memo_del > 0:
            base_date = today - timedelta(days=config.cf_memo_del)
            result = db.execute(
                delete(Memo).where(Memo.me_send_datetime < base_date)
            )
            print("删除备注 : ", base_date, f"{result.rowcount}删除枪")

        # 自动删除退出成员
        if config.cf_leave_day > 0:
            # TODO: 添加成员删除处理
            # query = update(Member).where(Member.mb_leave_date < datetime.now() - timedelta(days=config.cf_leave_day))
            # data = {}
            # result = db.execute(query, data)
            # print("会员删除参考日 : ", datetime.now() - timedelta(days=config.cf_leave_day), f"{result}删除枪")
            pass
        db.commit()
    except Exception as e:
        print(e)
    finally:
        db.close()


def is_possible_ip(request: Request, ip: str) -> bool:
    """IP允许访问 IP检查你的知识

    Args:
        request (Request): FastAPI Request 对象
        ip (str): IP

    Returns:
        bool: 全部允许 IP背面 True, 或者也许 False
    """
    cf_possible_ip = request.state.config.cf_possible_ip
    return check_ip_list(request, ip, cf_possible_ip, allow=True)


def is_intercept_ip(request: Request, ip: str) -> bool:
    """IP访问被阻止 IP检查你的知识

    Args:
        request (Request): FastAPI Request 对象
        ip (str): IP

    Returns:
        bool: 粗体 IP背面 True, 或者也许 False
    """
    cf_intercept_ip = request.state.config.cf_intercept_ip
    return check_ip_list(request, ip, cf_intercept_ip, allow=False)


def check_ip_list(request: Request, current_ip: str, ip_list: str, allow: bool) -> bool:
    """IP检查它是否属于特定列表的函数

    Args:
        request (Request): FastAPI Request 对象
        ip (str): IP
        ip_list (str): IP 列表字符串
        allow (bool): True如果允许列表, False阻止列表如果

    Returns:
        bool: 如果你属于名单， True, 或者也许 False
    """
    if request.state.is_super_admin:
        return allow

    ip_list = ip_list.strip()
    if not ip_list:
        return allow

    ip_patterns = ip_list.split("\n")
    for pattern in ip_patterns:
        pattern = pattern.strip()
        if not pattern:
            continue
        pattern = pattern.replace(".", r"\.")
        pattern = pattern.replace("+", r"[0-9\.]+")
        if re.match(f"^{pattern}$", current_ip):
            return True

    return False


def filter_words(request: Request, contents: str) -> str:
    """用于检查文章内容中是否存在已过滤单词的功能

    Args:
        request (Request): FastAPI Request 对象
        contents (str): 写作内容

    Returns:
        str: 如果有过滤的单词，对应的单词, 如果没有，则为空字符串
    """
    cf_filter = request.state.config.cf_filter
    words = cf_filter.split(",")
    for word in words:
        word = word.strip()
        if not word:
            continue
        if word in contents:
            return word

    return ''


def check_prohibit_words(request: Request, contents: str) -> str:
    """
    检查文章内容中是否有禁止使用的单词的功能

    Args:
        request (Request): FastAPI Request 对象
        contents (str): 写作内容

    Returns:
        str: 如果有一个禁止的词，对应的词, 如果没有，则为空字符串
    """
    prohibit_list = getattr(request.state.config, "cf_prohibit_id", "").split(",")
    prohibit_list = [id.strip().lower() for id in prohibit_list]

    if contents.lower() in prohibit_list:
        return contents

    return ''


def read_version():
    """根目录的 version.txt 读取文件以返回版本的函数
    Returns:
        str: 版本
    """
    with open("version.txt", "r", encoding="UTF-8") as file:
        return file.read().strip()


def read_license():
    """根目录的 LICENSE 读取文本文件以返回许可证内容

    Returns:
        str: 许可证内容
    """
    with open("LICENSE", "r", encoding="UTF-8") as file:
        return file.read().strip()


def get_current_admin_menu_id(request: Request) -> Optional[str]:
    """返回当前路径的管理员菜单想法的函数

    Args:
        request (Request): FastAPI Request 对象
        request (Request): FastAPI Request 객체

    Returns:
        Optional[str]: 经理菜单的想法
    """
    try:
        admin_menu = get_admin_menus()

        path = request.url.path
        routes = request.app.routes

        for route in routes:
            # 当前路径为 router 如果它符合格式
            if route.path_regex.match(path):
                tags = route.tags
                # 返回菜单想法以匹配路由器的标签
                for tag in tags:
                    for menu_items in admin_menu.values():
                        item = next((item for item in menu_items if item.get("tag", "") == tag), None)
                        if item:
                            return item.get("id")
                break

        # 基于路径的插件管理器搜索
        for route in routes:
            if route.path_regex.match(path):
                # 用户定义的管理员前缀访问复杂，因此删除后比较
                parts = route.path.split('/')
                modified_path = '/' + '/'.join(parts[2:])
                if result_menu_id := get_admin_menu_id_by_path(modified_path):
                    return result_menu_id

        raise Exception("“管理器” 菜单 “未找到任何想法”.")

    except Exception as e:
        logging.warning(e)
        return None


def remove_query_params(request: Request, keys: Union[str, list]) -> dict:
    """从查询参数中删除特定键.

    Args:
        request (Request): FastAPI Request 对象
        keys (Union[str, list]): 要删除的密钥

    Returns:
        dict: 查询参数
    """
    query_params_dict = dict(request.query_params)

    if isinstance(keys, str):
        keys = [keys]

    for key in keys:
        query_params_dict.pop(key, None)

    return query_params_dict


def set_url_query_params(url: Union[str, URL], query_params: Any) -> str:
    """包括查询参数 URL返回.

    Args:
        url (URL): URL 对象
        query_params (Any): 查询参数

    Returns:
        URL: 包括查询参数 URL
    """
    if isinstance(url, str):
        url = URL(url)

    return url.replace_query_params(**query_params).__str__()
from starlette.datastructures import URL

def safe_int_convert(string: str) -> int:
    """安全 int 变换函数"""
    try:
        return int(string)
    except (ValueError, TypeError):
        return 0


def get_paging_info(current_page: int, records_per_page: int, total_records: int) -> dict:
    """返回分页信息的函数

    Args:
        current_page (int): 当前页面
        records_per_page (int): 每页的记录数
        total_records (int): 记录的总数

    Returns:
        dict: 寻呼信息
    """

    offset = (current_page - 1) * records_per_page
    return {
        "offset": offset,
        "total_records": total_records,
        "current_page": current_page,
        "total_pages": math.ceil(total_records / records_per_page),
    }


def hide_ip_address(ip: str) -> str:
    """IP 撤消地址的函数"""
    return re.sub(r"([0-9]+)\.([0-9]+)\.([0-9]+)\.([0-9]+)",
                  "\\1.#.#.\\4", ip)
